package org.liziing.shortlink.project.mq.consumer;


import cn.hutool.core.date.DateUtil;
import cn.hutool.core.date.Week;
import cn.hutool.core.util.StrUtil;
import cn.hutool.http.HttpUtil;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import jodd.util.StringUtil;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.liziing.shortlink.project.common.convention.exception.ServiceException;
import org.liziing.shortlink.project.dao.entity.*;
import org.liziing.shortlink.project.dao.mapper.*;
import org.liziing.shortlink.project.dto.biz.ShortLinkStatsRecordDTO;
import org.liziing.shortlink.project.mq.idempotent.MessageQueueIdempotentHandler;
import org.liziing.shortlink.project.mq.producer.DelayShortLinkStatsProducer;
import org.redisson.api.RLock;
import org.redisson.api.RReadWriteLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.RecordId;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.stream.StreamListener;
import org.springframework.stereotype.Component;

import java.util.*;

import static org.liziing.shortlink.project.common.constant.RedisKeyConstant.LOCK_GID_UPDATE_KEY;
import static org.liziing.shortlink.project.common.constant.ShortLinkConstant.GET_LOCAL_URL;


/**
 * 短链接监控状态保存消息队列消费者
 */
@Slf4j
@Component
@RequiredArgsConstructor
public class ShortLinkStatsSaveConsumer implements StreamListener<String, MapRecord<String, String, String>> {

    private final ShortLinkMapper shortLinkMapper;
    private final ShortLinkGotoMapper shortLinkGotoMapper;
    private final RedissonClient redissonClient;
    private final LinkAccessStatsMapper linkAccessStatsMapper;
    private final LinkLocaleStatsMapper linkLocaleStatsMapper;
    private final LinkOsStatsMapper linkOsStatsMapper;
    private final LinkBrowserStatsMapper linkBrowserStatsMapper;
    private final LinkAccessLogsMapper linkAccessLogsMapper;
    private final LinkDeviceStatsMapper linkDeviceStatsMapper;
    private final LinkNetworkStatsMapper linkNetworkStatsMapper;
    private final LinkStatsTodayMapper linkStatsTodayMapper;
    private final DelayShortLinkStatsProducer delayShortLinkStatsProducer;
    private final StringRedisTemplate stringRedisTemplate;
    private final MessageQueueIdempotentHandler messageQueueIdempotentHandler;

    @Value("${short-link.local.restapi.amap-key}")
    private String amapKey;

    @Override
    public void onMessage(MapRecord<String, String, String> message) {
        String stream = message.getStream();
        RecordId id = message.getId();
        if (!messageQueueIdempotentHandler.idempotentMessageQueue(id.toString())) {
            if (messageQueueIdempotentHandler.hasAccomplish(id.toString())) {
                return;
            }
            throw new ServiceException("消息队列消费失败，请重试！");
        }
        try {
            Map<String, String> producerMap = message.getValue();
            String fullShortUrl = producerMap.get("fullShortUrl");
            if (StrUtil.isNotBlank(fullShortUrl)) {
                String gid = producerMap.get("gid");
                ShortLinkStatsRecordDTO statsRecord = JSON.parseObject(producerMap.get("statsRecord"), ShortLinkStatsRecordDTO.class);
                actualSaveShortLinkStats(fullShortUrl, gid, statsRecord);
            }
            // 直接删除
            stringRedisTemplate.opsForStream().delete(Objects.requireNonNull(stream), id.getValue());
        } catch (Throwable e) {
            messageQueueIdempotentHandler.delIdempotentMessageQueue(id.toString());
            log.debug("记录短链接监控消费异常！" + e);
        }
        messageQueueIdempotentHandler.setAccomplish(id.toString());
    }

    public void actualSaveShortLinkStats(String fullShortUrl, String gid, ShortLinkStatsRecordDTO statsRecord) {
        fullShortUrl = Optional.ofNullable(fullShortUrl).orElse(statsRecord.getFullShortUrl());
        RReadWriteLock readWriteLock = redissonClient.getReadWriteLock(String.format(LOCK_GID_UPDATE_KEY, fullShortUrl));
        RLock rLock = readWriteLock.readLock();
        if (!rLock.tryLock()) {
            delayShortLinkStatsProducer.send(statsRecord);
            return;
        }
        try {
            Date date = new Date();
            int hour = DateUtil.hour(date, true);
            Week weekday = DateUtil.dayOfWeekEnum(date);
            int week = weekday.getIso8601Value();
            if (StringUtil.isBlank(gid)) {
                LambdaQueryWrapper<ShortLinkGotoDO> queryWrapper = Wrappers.lambdaQuery(ShortLinkGotoDO.class)
                        .eq(ShortLinkGotoDO::getFullShortUrl, fullShortUrl);
                ShortLinkGotoDO shortLinkGotoDO = shortLinkGotoMapper.selectOne(queryWrapper);
                gid = shortLinkGotoDO.getGid();
            }
            LinkAccessStats linkAccessStats = LinkAccessStats.builder()
                    .pv(1)
                    .uv(statsRecord.getUvFirstFlag() ? 1 : 0)
                    .uip(statsRecord.getUipFirstFlag() ? 1 : 0)
                    .date(date)
                    .hour(hour)
                    .weekday(week)
                    .fullShortUrl(fullShortUrl)
                    .gid(gid)
                    .build();
            linkAccessStatsMapper.insertLinkAccessStatsUV(linkAccessStats);

            // =============================地区====================================
            Map<String, Object> localParam = new HashMap<>();
            localParam.put("key", amapKey);
            localParam.put("ip", statsRecord.getRemoteAddr());
            String localStr = HttpUtil.get(GET_LOCAL_URL, localParam);
            JSONObject localResultObj = JSON.parseObject(localStr);
            String infocode = localResultObj.getString("infocode");
            String realProvince = "未知";
            String realCity = "未知";
            if (!infocode.isBlank() && StrUtil.equals(infocode, "10000")) {
                String province = localResultObj.getString("province");
                boolean isProvinceExist = province.equals("[]");
                LinkLocaleStatsDO linkLocaleStatsDO = LinkLocaleStatsDO.builder()
                        .fullShortUrl(fullShortUrl)
                        .gid(gid)
                        .cnt(1)
                        .province(realProvince = isProvinceExist ? realProvince : province)
                        .city(realCity = isProvinceExist ? realCity : localResultObj.getString("city"))
                        .adcode(isProvinceExist ? "未知" : localResultObj.getString("adcode"))
                        .country("中国")
                        .date(date)
                        .build();
                linkLocaleStatsMapper.insertLinkLocaleStats(linkLocaleStatsDO);
            }

            LinkOsStatsDO linkOsStatsDO = LinkOsStatsDO.builder()
                    .os(statsRecord.getOs())
                    .fullShortUrl(fullShortUrl)
                    .gid(gid)
                    .cnt(1)
                    .date(date)
                    .build();
            linkOsStatsMapper.insertLinkOsStats(linkOsStatsDO);

            LinkBrowserStatsDO linkBrowserStatsDO = LinkBrowserStatsDO.builder()
                    .browser(statsRecord.getBrowser())
                    .fullShortUrl(fullShortUrl)
                    .gid(gid)
                    .cnt(1)
                    .date(date)
                    .build();
            linkBrowserStatsMapper.insertLinkBrowserStats(linkBrowserStatsDO);


            LinkDeviceStatsDO linkDeviceStatsDO = LinkDeviceStatsDO.builder()
                    .device(statsRecord.getDevice())
                    .cnt(1)
                    .gid(gid)
                    .fullShortUrl(fullShortUrl)
                    .date(date)
                    .build();
            linkDeviceStatsMapper.shortLinkDeviceState(linkDeviceStatsDO);
            LinkNetworkStatsDO linkNetworkStatsDO = LinkNetworkStatsDO.builder()
                    .network(statsRecord.getNetwork())
                    .cnt(1)
                    .gid(gid)
                    .fullShortUrl(fullShortUrl)
                    .date(date)
                    .build();
            linkNetworkStatsMapper.shortLinkNetworkState(linkNetworkStatsDO);

            //统计高频ip
            LinkAccessLogsDO linkAccessLogsDO = LinkAccessLogsDO.builder()
                    .ip(statsRecord.getRemoteAddr())
                    .os(statsRecord.getOs())
                    .browser(statsRecord.getBrowser())
                    .gid(gid)
                    .fullShortUrl(fullShortUrl)
                    .user(statsRecord.getUv())
                    .device(statsRecord.getDevice())
                    .network(statsRecord.getNetwork())
                    .locale(StrUtil.join("-", "中国", realProvince, realCity))
                    .build();
            linkAccessLogsMapper.insert(linkAccessLogsDO);

            shortLinkMapper.incrementStats(gid, fullShortUrl, 1, statsRecord.getUvFirstFlag() ? 1 : 0, statsRecord.getUipFirstFlag() ? 1 : 0);

            Long addUv = stringRedisTemplate.opsForSet().add("short-link:stats:uv:" + fullShortUrl + ":" + DateUtil.formatDate(date), statsRecord.getUv());
            boolean uvFirstTodayFlag = addUv != null && addUv > 0L;
            Long addUip = stringRedisTemplate.opsForSet().add("short-link:stats:uv:" + fullShortUrl + ":" + DateUtil.formatDate(date), statsRecord.getRemoteAddr());
            boolean uipFirstTodayFlag = addUip != null && addUip > 0L;

            // 统计今日
            LinkStatsTodayDO linkStatsTodayDO = LinkStatsTodayDO.builder()
                    .gid(gid)
                    .fullShortUrl(fullShortUrl)
                    .date(date)
                    .todayPv(1)
                    .todayUv(uvFirstTodayFlag ? 1 : 0)
                    .todayUip(uipFirstTodayFlag ? 1 : 0)
                    .build();
            linkStatsTodayMapper.insertLinkTodayStats(linkStatsTodayDO);
        } catch (Throwable ex) {
            log.error("短链接访问量统计异常", ex);
        }finally {
            rLock.unlock();
        }
    }
}
